package com.hhf.rrd.usage;

import com.hhf.rrd.model.Access;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * table api
 *      问题：SlotNotFoundException: Could not find slot for a726ecfe0d6afcb3c47a7b857c6de624.
 *
 * @author huanghaifeng15
 * @date 2022/2/11 13:43
 **/
public class TableSqlApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStreamSource = env.readTextFile("data/access.log");
        SingleOutputStreamOperator<Access> streamOperator = dataStreamSource.map(new RichMapFunction<String, Access>() {
            @Override
            public Access map(String value) throws Exception {
                String[] wordArr = value.split(",");
                Long time = Long.parseLong(wordArr[0].trim());
                String domain = wordArr[1].trim();
                Double traffic = Double.parseDouble(wordArr[2].trim());

                return new Access(time, domain, traffic);
            }
        });
        streamOperator.print("map: ");

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Table table = tableEnv.fromDataStream(streamOperator);

        tableEnv.createTemporaryView("access",table);

        Table resultTable = tableEnv.sqlQuery("select domain, sum(traffic) as traffics from access group by domain");
        tableEnv.toRetractStream(resultTable, Row.class).filter(new FilterFunction<Tuple2<Boolean, Row>>() {
            @Override
            public boolean filter(Tuple2<Boolean, Row> value) throws Exception {
                return value.f0;
            }
        }).print("row:0 ");

        Table resultTable1 = tableEnv.sqlQuery("select * from access where domain='jd.com'");
        tableEnv.toRetractStream(resultTable1,Row.class).filter(x -> x.f0).print("row:1 ");

        Table resultTable2 = table.select("*").where("domain='jd.com'");
        tableEnv.toAppendStream(resultTable2, Row.class).print("row:2 ");

        Table resultTable3 = table.select($("domain"),$("traffic"));
        tableEnv.toAppendStream(resultTable3, Row.class).print("row:3 ");

        env.execute();
    }
}
